/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package users;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;

import javax.jms.JMSException;
import javax.jms.TopicSubscriber;

import org.apache.qpid.AMQException;
import org.apache.qpid.client.message.JMSTextMessage;
import org.apache.qpid.url.URLSyntaxException;

/**
 * Publisher to a Qpid broker.
 */
public class Publisher extends User {

    /** List of topics to publish to. */
    private Set<String> topics = new HashSet<String>();
    /**
     * Getter for the variable 'topics'.
     * @return variable 'topics'
     */
    public Set<String> getTopics() {
        return topics;
    }
    /**
     * Setter for the variable 'topics'.
     * @param topics  new value of the variable 'topics'
     */
    public void setTopics(final Set<String> topics) {
        this.topics = topics;
    }

    /** Topics mapped to their primary connection. */
    private HashMap<String, PublisherConnection> topicPrims
            = new HashMap<String, PublisherConnection>();
    /** Topics mapped to their mirror connection. */
    private HashMap<String, PublisherConnection> topicMirrors
            = new HashMap<String, PublisherConnection>();
    /**
     * Getter for the variable 'topicPrims'.
     * @return variable 'topicPrims'
     */
    public HashMap<String, PublisherConnection> getTopicPrims() {
        return topicPrims;
    }
    /**
     * Setter for the variable 'topicPrims'.
     * @param topicPrims  new value of the variable 'topicPrims'
     */
    public void setTopicPrims(final HashMap<String,
                                  PublisherConnection> topicPrims) {
        this.topicPrims = topicPrims;
    }
    /**
     * Getter for the variable 'topicMirrors'.
     * @return variable 'topicMirrors'
     */
    public HashMap<String, PublisherConnection> getTopicMirrors() {
        return topicMirrors;
    }
    /**
     * Setter for the variable 'topicMirrors'.
     * @param topicMirrors  new value of the variable 'topicMirrors'
     */
    public void setTopicMirrors(final HashMap<String,
                                PublisherConnection> topicMirrors) {
        this.topicMirrors = topicMirrors;
    }


    /**
     * Empty constructor with no parameters.
     */
    public Publisher() {
        //after retrieving a publisher from a database, its scAssigTopic,
        //scUpdatesTopic and omUpMsgListener MUST be set manually
    }

    /**
     * Constructor for new publishers.
     *
     * Opens a connection and creates a session to the broker to publish
     * messages to.
     *
     * @param username  username to use to connect to the broker
     * @param password  password to use to connect to the broker
     * @param topics    list of topics to publish to
     */
    public Publisher(final String username, final String password,
                     final Set<String> topics) {

        super(username, password);

        this.topics = topics;

        //get the brokers to publish to and the PublisherConnections for them
        requestBroker();

    }


    /**
     * Add a topic to the list of topics to publish about and create publishers
     * for it in both brokers.
     *
     * @param topicName  the name of the topic to be added
     * @return  the name of the topic added or null if there is any problem
     */
    public String addTopic(final String topicName) {

        //if a topic with that name doesn't already exist
        if (!getTopics().contains(topicName)) {
            //add the topic to the lists of topics
            getTopics().add(topicName);

            //ask the overlay manager what broker to use
            try {
                //send the message to the overlay manager
                String msgText = "broker request:publisher:" + getClientId()
                                 + ":" + topicName;
                getOmConnection().sendMessage(getOmAssigPub(), msgText);

                System.out.println("Publisher " + getClientId()
                        + " waiting for a broker to publish to: " + msgText);

                //wait until a message is received
                JMSTextMessage recMsg
                        = (JMSTextMessage) getOmAssigSub().receive();

                //analyse the text of the received message
                String text = recMsg.getText();
                String[] parts = text.split(":");
                if (parts.length == NUM_PARTS_ASSIG_MSG) {
                    if (parts[PB_INDEX_ASSIG].equals("null")) {
                        throw new Exception("No primary broker assigned");
                    }
                    //add topic to connection with primary broker
                    PublisherConnection primConn
                            = addTopicToConn(topicName, parts[PB_INDEX_ASSIG],
                                    Integer.parseInt(parts[PP_INDEX_ASSIG]));
                    //add entry to map topics-primary connections
                    getTopicPrims().put(topicName, primConn);

                    if (parts[MB_INDEX_ASSIG].equals("null")) {
                        throw new Exception("No mirror broker assigned");
                    }
                    //add topic to connection with mirror broker
                    PublisherConnection mirrorConn
                            = addTopicToConn(topicName, parts[MB_INDEX_ASSIG],
                                    Integer.parseInt(parts[MP_INDEX_ASSIG]));
                    //and entry to map topics-mirror connections
                    getTopicMirrors().put(topicName, mirrorConn);

                    System.out.println("Publisher " + getClientId()
                            + " has been assigned primary broker "
                            + primConn.getBroker() + " and primary port "
                            + primConn.getPort() + " for topic " + topicName);
                    System.out.println("Publisher " + getClientId()
                            + " has been assigned mirror broker "
                            + mirrorConn.getBroker() + " and mirror port "
                            + mirrorConn.getPort() + " for topic "
                            + topicName);

                    return topicName;
                } else {
                    throw new Exception("Error assigning brokers: " + text);
                }

            } catch (Exception e) {
                System.err.println(e.getMessage());
            }

        }
        return null;

    }

    /**
     * Publish a text message to primary and mirror brokers of a topic.
     *
     * Messages are published first to the mirror broker so the timestamp of a
     * message in the mirror broker is always smaller than the timestamp of the
     * corresponding message in the primary broker. Mirror subscribers can use
     * this to decide whether they should receive a message or not.
     *
     * @param topicName  topic to send the message to
     * @param text  text of the message
     * @param properties  properties of the message
     * @throws JMSException  if there is any problem
     */
    public void publishMessage(final String topicName, final String text,
            final HashMap<String, String> properties)  throws JMSException {

        //get a correlation ID for the message
        String corrID = UUID.randomUUID().toString();

        while (!getTopicPrims().get(topicName).isConnected()
               || !getTopicMirrors().get(topicName).isConnected()) {
            //wait until both primary and mirror connections are ready
        }

        //publish the message to the mirror broker
        getTopicMirrors().get(topicName).publishMessage(
                topicName, text, properties, corrID);
        //publish the message to the primary broker
        getTopicPrims().get(topicName).publishMessage(
                topicName, text, properties, corrID);

    }

    /**
     * Add a topic to the connection with the given broker.
     * Create the connection if it doesn't exist.
     *
     * @param topicName  name of the topic to add
     * @param broker  IP address of the broker
     * @param port  port of the broker
     * @return  connection in which the topic has been added
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     * @throws URLSyntaxException  if there is any problem
     */
    private PublisherConnection addTopicToConn(final String topicName,
            final String broker, final int port)
            throws JMSException, URLSyntaxException, AMQException {

        //get a connection with the broker
        PublisherConnection conn
                = (PublisherConnection) getConnForBroker(broker);
        if (conn == null) {
            conn = new PublisherConnection(this, broker, port,
                                           getMOM_EXCHANGE());
            conn.initialize();
            //add the primary connection to the list of connections
            getConnections().add(conn);
        }
        //add the topic to the connection
        conn.addTopic(topicName);

        return conn;

    }

    /**
     * When a publisher is created, it asks the overlay manager what brokers
     * it has to use sending a message to the overlay manager.
     *
     * The format of the message is:
     *
     * 'broker request:publisher:<clientId>:<topic1>:<topic2>:...'
     *
     * The received message will have the format:
     *
     * 'topic1:primaryBroker1:primaryPort1:mirrorBroker1:mirrorPort1:topic2:...'
     */
    @Override
    public void requestBroker() {

        try {
            //initialize connection with the overlay manager
            getOmConnection().initialize();

            //create a subscriber to messages from the overlay manager
            setOmAssigSub(getOmConnection().getSubscriberForTopic(
                    getOmAssigTopic()));

            //create a publisher to send messages to the overlay manager
            setOmAssigPub(
                    getOmConnection().getPublisherForTopic(getOM_TOPIC()));
            
            if (!getTopics().isEmpty()) {
                //send the message to the overlay manager
                String msgText = "broker request:publisher:" + getClientId();
                for (String topic : getTopics()) {
                    msgText = msgText + ":" + topic;
                }
                getOmConnection().sendMessage(getOmAssigPub(), msgText);
                //wait until a message is received
                System.out.println("Publisher " + getClientId()
                        + " waiting for being assigned a broker");
                JMSTextMessage recMsg = (JMSTextMessage) getOmAssigSub()
                        .receive();
                //analyse the text of the received message
                String text = recMsg.getText();
                String[] parts = text.split(":");
                if ((parts.length % NUM_PARTS_ASSIG_MSG) == 0) {
                    for (int i = 0;
                         i < (parts.length - (NUM_PARTS_ASSIG_MSG - 1));
                         i += NUM_PARTS_ASSIG_MSG) {
                        String topicName = parts[i]; //topic

                        if (parts[i + PB_INDEX_ASSIG].equals("null")) {
                            throw new Exception("No primary broker assigned");
                        }
                        //add topic to connection with the primary broker
                        PublisherConnection primConn = addTopicToConn(
                                topicName, parts[i + PB_INDEX_ASSIG], Integer
                                        .parseInt(parts[i + PP_INDEX_ASSIG]));
                        //add entry to map topics-primary connections
                        getTopicPrims().put(topicName, primConn);

                        if (parts[i + MB_INDEX_ASSIG].equals("null")) {
                            throw new Exception("No mirror broker assigned");
                        }
                        //add topic to connection with the mirror broker
                        PublisherConnection mirrorConn = addTopicToConn(
                                topicName, parts[i + MB_INDEX_ASSIG], Integer
                                        .parseInt(parts[i + MP_INDEX_ASSIG]));
                        //add entry to map topics-mirror connections
                        getTopicMirrors().put(topicName, mirrorConn);

                        System.out.println("Publisher " + getClientId()
                                + " has been assigned primary broker "
                                + primConn.getBroker() + " and primary port "
                                + primConn.getPort() + " for topic "
                                + topicName);
                        System.out.println("Publisher " + getClientId()
                                + " has been assigned mirror broker "
                                + mirrorConn.getBroker() + " and mirror port "
                                + mirrorConn.getPort() + " for topic "
                                + topicName);
                    }

                } else {
                    throw new Exception("Error assigning brokers: " + text);
                }
            }
            //create a subscriber for updates from the overlay manager
            TopicSubscriber subscriber = getOmConnection()
                    .getSubscriberForTopic(getOmUpdatesTopic());
            //set a message listener for updates
            subscriber.setMessageListener(getOmUpMsgListener());

        } catch (Exception e) {
            System.err.println("It hasn't been possible to assign a broker");
            e.printStackTrace();
        }
    }

    /**
     * Change the primary broker of a publisher for a topic.
     * The old mirror broker of the publisher becomes its new primary broker and
     * the given broker becomes its new mirror broker.
     *
     * @param topic  topic to change
     * @param newBroker  new mirror broker to use
     * @param newPort  new mirror port to use
     * @throws URLSyntaxException  if there is any problem
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     * @throws Exception  if no new mirror broker is assigned
     */
    protected void changePrimary(final String topic, final String newBroker,
            final int newPort)
            throws URLSyntaxException, JMSException, AMQException, Exception {

        //connection of the topic
        PublisherConnection oldPrimCon = getTopicPrims().get(topic);
        try {
            //remove topic from old primary connection
            oldPrimCon.removeTopic(topic);
            //if the connection is closed, remove it
            if (!oldPrimCon.isConnected()) {
                disconnect(oldPrimCon);
            }
        } catch (Exception e) {
            //connection didn't exist or was already closed
            disconnect(oldPrimCon);
        }

        //mirror connection becomes primary connection
        getTopicPrims().put(topic, getTopicMirrors().get(topic));
        //TODO: for a short time, primary and mirror are the same connection
        if (newBroker.equals("null")) {
            throw new Exception("No mirror broker assigned");
        }
        //add topic to connection with mirror broker
        PublisherConnection mirrorConn
                = addTopicToConn(topic, newBroker, newPort);
        //add entry to map topics-mirror connections
        getTopicMirrors().put(topic, mirrorConn);

        System.out.println("Publisher " + getClientId()
                + " has been reassigned primary broker "
                + getTopicPrims().get(topic).getBroker()
                + " and primary port " + getTopicPrims().get(topic).getPort()
                + " for topic " + topic);
        System.out.println("Publisher " + getClientId()
                + " has been reassigned mirror broker "
                + getTopicMirrors().get(topic).getBroker()
                + " and mirror port " + getTopicMirrors().get(topic).getPort()
                + " for topic " + topic);
    }

    /**
     * Change the mirror broker of a publisher for a topic.
     * The given broker becomes the new mirror broker of the publisher.
     *
     * @param topic  topic to change
     * @param newBroker  new mirror broker to use
     * @param newPort  new port to use
     * @throws URLSyntaxException  if there is any problem
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     * @throws Exception  if no new mirror broker is assigned
     */
    protected void changeMirror(final String topic, final String newBroker,
            final int newPort)
            throws URLSyntaxException, JMSException, AMQException, Exception {

        //connection of the topic
        PublisherConnection oldMirrorCon = getTopicMirrors().get(topic);
        try {
            //remove topic from old mirror connection
            oldMirrorCon.removeTopic(topic);
            //if the connection is closed, remove it
            if (!oldMirrorCon.isConnected()) {
                disconnect(oldMirrorCon);
            }
        } catch (Exception e) {
            //connection didn't exist or was already closed
            disconnect(oldMirrorCon);
        }

        if (newBroker.equals("null")) {
            throw new Exception("No mirror broker assigned");
        }
        //add topic to connection with new mirror broker
        PublisherConnection mirrorConn
                = addTopicToConn(topic, newBroker, newPort);
        //add entry to map topics-mirror connections
        getTopicMirrors().put(topic, mirrorConn);

        System.out.println("Publisher " + getClientId()
                + " has been reassigned mirror broker "
                + getTopicMirrors().get(topic).getBroker()
                + " and mirror port " + getTopicMirrors().get(topic).getPort()
                + " for topic " + topic);
    }

}
